package cn.doitedu.udf

import cn.doitedu.util.SparkUtil
import org.apache.spark.sql.types.DataType
import org.apache.spark.sql.{DataFrame, Dataset, Row}

/**
 * @Date 22.4.13
 * @Created by HANGGE
 * @Description
 */
object C02_Demo02 {
  def main(args: Array[String]): Unit = {
    val session = SparkUtil.getSession
     import  session.implicits._
    import  org.apache.spark.sql.functions._
    // 加载数据
    val df =session.read.option("header" , true).option("inferschema" ,true).csv("data\\udf\\user.csv")
     // 处理数据 将参与比对的向量存储在数组中
    println("---------------------------------------------------")
    // 推荐使用的方式
     val df2: DataFrame = df.map(row => {
       // val id = row.getInt(0)
       val id = row.getAs[Int]("id")
       val name = row.getAs[String]("name")
       // 注意数据解析的实际类型
       val age = row.getAs[Int]("age").toDouble
       val height = row.getAs[Int]("height").toDouble
       val weight = row.getAs[Int]("weight").toDouble
       val yanzhi = row.getAs[Int]("yanzhi").toDouble
       val score = row.getAs[Double]("score")
       (id, name, Array[Double](age, height, weight, yanzhi, score))
     }).toDF("id", "name", "featrues")

   // df2.show(false)
    println("---------------------------------------------------")
    // 处理方式02
    df.map({
      case Row(id:Int , name:String , age:Int , height:Int , weight:Int , yanzhi:Int , score:Double)
      => (id,name ,Array[Double](age,height , weight ,yanzhi , score))
    }).toDF("id", "name", "featrues")
    println("---------------------------------------------------")
    // 处理方式03  select  查询字段  类型转换
   /* df.select("id" , "name" , "array(age , heigth)")
    df.select('id ,'name , array('age.cast("double") , 'height)) //类型转换*/

    println("---------------------------------------------------")

  // 处理方式04  样例类
    df.map({
      case Row(id:Int , name:String , age:Int , height:Int , weight:Int , yanzhi:Int , score:Double)
      => User(id,name ,Array[Double](age,height , weight ,yanzhi , score))
    })

    println("====================第二步==============================")
    val joined = df2.join(df2.toDF("id2", "name2", "featrues2"), 'id < 'id2)
    joined.show(100 ,false)
    println("====================第三步==============================")
    // 自定义函数
        // 普通的数组在反复调用复制的时候出问题  增强的数组
  /*  (arr1:Array[Double]  ,arr2:Array[Double])=>{
      0.0d
    }*/
    import scala.collection._
    // 必须是可变的是数组
   val cos_sime: (Array[Double], Array[Double]) => Double = (arr1:Array[Double], arr2:Array[Double])=>{
     val fm01 = Math.pow(arr1.map(Math.pow(_, 2)).sum, 0.5)
     val fm02 = Math.pow(arr2.map(Math.pow(_, 2)).sum, 0.5)
     val fz = arr1.zip(arr2).map(tp => (tp._1 * tp._2)).sum
     fz / (fm01*fm02)
    }
    println("====================第四步==============================")
    // 注册函数
    session.udf.register("cos_sime" , cos_sime)
    joined.createOrReplaceTempView("tb_user")
    session.sql(
      """
        |select
        |id ,
        |name ,
        |id2 ,
        |name2  ,
        |cos_sime(featrues, featrues2) as cos
        |from
        |tb_user
        |""".stripMargin).show()
    // 注册方式02    使用DSL风格编写代码
    val myfunc  = udf(cos_sime)
    joined.select('id , 'id2 , 'name , 'name2 , myfunc('featrues ,'featrues2) as "res").show()

    /**
     * root
     * |-- id: integer (nullable = true)
     * |-- name: string (nullable = true)
     * |-- age: integer (nullable = true)   -
     * |-- height: integer (nullable = true) -
     * |-- weight: integer (nullable = true) -
     * |-- yanzhi: integer (nullable = true) -
     * |-- score: double (nullable = true)   -
     */
    df.printSchema()




  }

}
